package com.burges.net.tableAPIAndSQL.table.api

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.java.{Over, Slide}
import org.apache.flink.table.api.scala.{Session, Tumble}
import org.apache.flink.table.api.{Table, TableEnvironment, Window}
import sun.plugin2.jvm.CircularByteBuffer.Streamer

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/15 
  * 描述     flink table api应用实例
  */
object FlinkTableAPIDEMO {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val batchEnv = ExecutionEnvironment.getExecutionEnvironment
		//获取TableEnvironment
		val tStreamEnv = TableEnvironment.getTableEnvironment(environment)
		val tBatcbEnv = TableEnvironment.getTableEnvironment(batchEnv)

		val stream1: DataStream[(String, String)] = environment.fromElements(("a", "b"))
		val stream2: DataStream[(String, String)] = environment.fromElements(("a", "b"))
		val stream3: DataSet[(String, String)] = batchEnv.fromElements(("a", "b"))
		val stream4: DataSet[(String, String)] = batchEnv.fromElements(("a", "b"))
		
		//通过scan方法在catalog中找到Sensors表
		val sensors: Table = tStreamEnv.scan("Sensors")
		//对该表使用TableAPI进行处理
		/**
		  * 查询和过滤
		  */
		val rs = sensors
        		.groupBy("id") //根据id进行GroupBy操作
				// 使用select方法查询指定字段，并通过as进行字段重命名
        		.select("id", 'var1.sum as 'val1Sum) //查询id和var1的sum指标
        		.toAppendStream[(String, Long)]    //将处理结果转换成元祖类型DataStream数据集

		// 使用select*将所有的字段查询出来
		val rs2 = sensors.select("*")
		// 使用as方法进行重命名
		val sensorsAs:Table = tStreamEnv.scan("Sensors").as('a, 'b, 'c, 'd)

		// 使用filter方法进行数据筛选
		sensors.filter('var1%2 ===0)
		// 使用where方法进行数据筛选
		sensors.where('id === "1001")

		/**
		  * 窗口操作
		  */
		var rs3 = sensors.window([w: Window] as 'window)
        		.groupBy('window', 'id)
        		.select('id, 'var1.sum, 'window.start, 'window.end, 'window.rowtime)
		//=======================================================
		//窗口类型GroupByWindow
		//=======================================================
		// 1.Tumbling Windows滚动窗口
		// over操作符指定窗口的长度
		// on操作符定义了窗口基于的时间概念类型为EventTime还是ProcessTime
		// as操作符将创建的窗口进行重命名，同时名称需要在后续的算子中使用
		//基于EventTime时间概念创建滚动窗口，窗口长度为1h
		sensors.window(Tumble over 1.hour on 'rowtime as 'window)
		//基于Processime时间概念创建滚动窗口，窗口长度为1h
		sensors.window(Tumble over 1.hour on 'proctime as 'window)
		//基于EventTime时间概念创建滚动窗口，窗口长度为100条记录
		sensors.window(Tumble over 1.rows on 'rowtime as 'window)

		// 2.SlidingWindows
		// over操作符指定窗口的长度
		// on操作符定义了窗口基于的时间概念类型为EventTime还是ProcessTime
		// every操作符指定窗口的移动频率
		// as操作符将创建的窗口进行重命名，同时名称需要在后续的算子中使用
		//基于EventTime时间概念创建滑动窗口，窗口长度为10分钟，每隔5s统计一次
		sensors.window(Slide over 10.minutes every 5.mills on 'rowtime as 'window)
		//基于Processime时间概念创建滑动窗口，窗口长度为10分钟，每隔5s统计一次
		sensors.window(Slide over 10.minutes every 5.mills on 'proctime as 'window)
		//基于EventTime时间概念创建滑动窗口，窗口长度为10分钟，每隔5s统计一次，proctime没有实际意义
		sensors.window(Slide over 100.rows every 5.rows on 'rowtime as 'window)

		// 3.SessionWindows
		// 基于EventTime时间概念创建会话窗口，SessionGap为10mins
		sensors.window(Session withGap 10.minutes on 'rowtime as 'window)
		// 基于ProcessTime时间概念创建会话窗口，SessionGap为10mins
		sensors.window(Session withGap 10.minutes on 'proctime as 'window)

		//=======================================================
		//窗口类型OverWindow
		//=======================================================
		// 创建依赖于partitionBy、orderBy、preceding及following四个参数
		// partitionBy操作符指定了一个或多个分区字段，根据指定的字段进行分区处理，算子运算，可选项，如果不指定，那么会再一个Task实例中完成计算
		// orderBy操作符制定了数据排序的字段，通常情况下使用EventTime或ProcessTime进行时间排序
		// preceding操作符指定基于当前数据需要向前纳入多少数据作为窗口的范围，有两种类型Bounded类型和UnBounded类型
		//   Bounded类型指定当前数据之前的多少条数据
		//   UnBounded类型表示从进入系统的第一条数据开始
		//       使用静态变量UNBOUNDED_RANGE指定，表示以时间为单位的数据范围
		//       使用UNBOUNDED_ROW使用，使用以数据量为单位的数据范围
		// following操作符指定从当前记录开始向后纳入多少数据作为计算的范围，目前还不支持
		//   可以使用静态变量CURRENT_ROW和CURRENT_RANGE来设定仅包含当前行
		//   默认情况系Flink会根据用户使用窗口间隔是时间还是数量来指定following参数
		//   指定的单位必须和preceding的单位一直
		sensors
				//指定OverWindow并重命名为window
        		.window(Over partitionBy 'id orderBy 'rowTime preceding UNBOUNDED_RANGE as 'window)
				//通过在Select操作符中指定查询字段，窗口上var1求和值和var2的最大值
        		.select('id, 'var1.sum over 'window, 'var2.max over 'window)
		sensors
				//指定OverWindow并重命名为window preceding 10分钟
				.window(Over partitionBy 'id orderBy 'rowTime preceding 10min as 'window)
				//通过在Select操作符中指定查询字段，窗口上var1求和值和var2的最大值
				.select('id, 'var1.sum over 'window, 'var2.max over 'window)
		sensors
				//指定OverWindow并重命名为window preceding 100条记录
				.window(Over partitionBy 'id orderBy 'rowTime preceding 100.rows as 'window)
				//通过在Select操作符中指定查询字段，窗口上var1求和值和var2的最大值
				.select('id, 'var1.sum over 'window, 'var2.max over 'window)

		/**
		  * 聚合操作
		  */
		// GroupBy Aggregation
		// 根据ID进行聚合，求取var1字段的sum结果
		sensors.groupBy('id).select('id, 'var.sum as 'var1Sum)
		// GroupBy Window Aggregation
		sensors
				// 定义窗口类型为滚动窗口
        		.window(Tumble over 1.hour on 'rowtime as 'window)
				// 根据id和window进行聚合
        		.groupBy('id, 'window)
				// 根据字段id 窗口属性 以及聚合指标var1sum
        		.select('id, 'window.start, 'window.end, 'var1.sum as 'vars1Sum)
		// Over Window Aggregation
		sensors
				//指定OverWindow并重命名为window
				.window(Over partitionBy 'id orderBy 'rowTime preceding UNBOUNDED_RANGE as 'window)
				//通过在Select操作符中指定查询字段，窗口上var1求和值和var2的最大值
				.select('id, 'var1.sum over 'window, 'var2.max over 'window)
		// Distinct Aggregation
		sensors
				.window(Tumble over 1.minutes on 'rowtime as 'window)
        		.groupBy('id, 'window)
        		.select('id, 'var1.sum.distinct as 'var1Sum)
		// distinct
		var distinctRs = sensors.distinct()

		/**
		  * 多表联查
		  */
		// Inner Join
		val t1 = tStreamEnv.fromDataStream(stream1, 'id1, 'var1, 'var2)
		val t2 = tStreamEnv.fromDataStream(stream2, 'id2, 'var3, 'var4)
		val innseJoinRs = t1.join(t2).where('id1 === 'id2).select('id1, 'var1, 'var3)
		// Outer Join
		// left outer join
		var leftOuterRs = t1.leftOuterJoin(t2, 'id1 === 'id2).select('id1, 'var1, 'var3)
		// right outer join
		var rightOuterRs = t1.rightOuterJoin(t2, 'id1 === 'id2).select('id1, 'var1, 'var3)
		// full outer join
		var fullOuterRs = t1.fullOuterJoin(t2, 'id1 === 'id2).select('id1, 'var1, 'var3)
		// Time-windowed Join
		val t3 = tBatcbEnv.fromDataSet(stream3, 'id1, 'var1, 'var2, 'time1.rowtime)
		val t4 = tBatcbEnv.fromDataSet(stream4, 'id2, 'var3, 'var4, 'time2.rowtime)
		// 将t3和t4表关联，并在where操作符中指定时间关联条件
		var rsTimeWindowedJoin = t3.join(t4).where('id1 === 'id2 && 'time1 >= 'time2 - 10.minutes && 'time1 < 'time2 +10.minutes)
        		.select('id1, 'var1, 'var2, 'time1)
		// Join with Table Function
		var upper: TableFunction[_] = new MyUpperUDTF()
		// 通过InnerJoin关联经过MyUpperUDTF处理的Table，然后形成新的表
		var result: Table = table.join(upper('var1) as 'upperVar1).select('id, 'var1, 'upperVar1, 'var2)
		// Join with Temporal Table
		val tempTable = tStreamEnv.scan("TempTable")
		val temps = tempTable.createTemporalTableFunction('t_protime, 't_id)
		val table = tStreamEnv.scan("Table")
		val rs = table.join(temps('o_rowtime), 'table_key == 'temp_key)

		/**
		  * 集合操作
		  */
		val unionTable = t1.union(t2)
		val unionAllTable = t1.unionAll(t2)
		val intersectTable = t1.intersect(t2)
		val intersectAllTable = t1.intersectAll(t2)
		val minusTable = t1.minus(t2)
		val munusAllTable = t1.minusAll(t2)
		val result1 = t1.where('id in(right))
		val result2 = t1.where('id in ("92", "11"))

		/**
		  * 排序操作
		  */
		// asc desc
		// fetch输出的条数
		// offset输出的偏移量
		t3.orderBy('var1.asc).fetch(5).offset(10)

		/**
		  * 数据写入
		  */
		sensors.insertInto("OutSensros")
	}


}
